#ifndef __M_MSG_H__
#define __M_MSG_H__


// 消息的管理
// 消息的要素：
// 1,网络传输的消息要素:  消息属性(消息id，消息routing_key, 消息的投递模式)  &&实际内容(数据)
// 2,服务器上的消息管理所需要的额外要素:最主要的就是持久化管理
//  1,消息有效标志
//  每一条消息都有可能进行持久化存储,等到推送给客户端就会删除掉
//  然而每次删除一条数据就就重写一次文件,效率太低下了。
//  如果设置了有效标志位，每次只需要将这个有效标志位对应的数据给修改为无效就行
//  消息的实际存储位置(相对于文件起始位置的偏移量)

// 队列数据管理
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <list>

// 消息的属性
//  message BasicProperties
//  {
//      string id = 1;  //消息id
//      DeliveryMode delivery_mode = 2;
//      string routing_key = 3;  //
//  };
//  message Message
//  {
//      message Payload
//      {
//          BasicProperties properties = 1; //消息的属性
//          string body = 2; //有效载荷数据
//          string valid = 3; //消息是否有效
//      };

//     Payload payload = 1; //
//     uint32 offset = 2; // 在持久化文件中的位置
//     uint32 length =3;  // 长度
// };

namespace chenmq
{
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"

    using MessagePtr = std::shared_ptr<chenmq::Message>;
    class MessageMapper
    {
    public:
        MessageMapper(std::string &basedir, const std::string &qname)
            : _qname(qname)
        {
            if (basedir.back() != '/')
                basedir.push_back('/');

            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;

            // 不存在则创建对应的目录
            if (FileHelper(basedir).exists() == false)
                assert(FileHelper::createDirectory(basedir));

            createMsgFile();
        }

        bool createMsgFile()
        {
            if (FileHelper(_datafile).exists() == true)
                return true;

            bool ret = FileHelper::createFile(_datafile);
            if (ret == false)
            {
                DLOG("创建队列数据 %s 失败!", _datafile.c_str());
                return false;
            }

            return true;
        }

        // 删除数据文件和临时文件
        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }

        // 插入一条消息message
        bool insert(MessagePtr &msg)
        {
            return insert(_datafile, msg);
        }

        bool insert(const std::string &filename, MessagePtr &msg)
        {
            // 新增数据需要添加在文件的末尾

            // 其中payload数据是消息属性(id, 是否持久化，routing_key)， 消息正文，消息是否有效
            std::string body = msg->payload().SerializeAsString();
            // 获取文件的长度
            FileHelper helper(filename);
            size_t fsize = helper.size();
            size_t msg_size = body.size();
            // 写入逻辑: 1,先写入4个字节的数据长度  2,再写入指定长度的数据
            bool ret = helper.write((char *)&msg_size, fsize, sizeof(size_t));
            if (ret == false)
            {
                DLOG("向队列数据文件写入消息长度失败");
                return false;
            }
            ret = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
            if (ret == false)
            {
                DLOG("向队列数据文件写入消息数据失败");
                return false;
            }

            msg->set_offset(fsize + sizeof(size_t)); // 这个是消息数据在文件中的偏移量
            msg->set_length(body.size());            // 这个是设置消息的大小
            return true;
        }

        bool remove(MessagePtr &msg)
        {
            // 1, 将msg中的有效标志位修改为 '0'
            msg->mutable_payload()->set_valid("0");
            // 2, 对msg进行序列化
            std::string body = msg->payload().SerializeAsString();
            if (body.size() != msg->length())
            {
                DLOG("无法修改文件中的数据信息, 因为新生成的数据与原数据长度不一致");
                return false;
            }

            // 3,将序列化后的消息，写入到数据在文件中的指定位置 (覆盖原来有的数据)
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), msg->offset(), body.size());
            if (ret == false)
            {
                DLOG("向队列数据写入数据失败!");
                return false;
            }

            return true;
        }

        std::list<MessagePtr> gc()
        {
            // 1,先从文件中加载有效消息
            bool ret;
            std::list<MessagePtr> result;
            ret = load(result);
            if (ret == false)
            {
                DLOG("加载有效数据失败\n");
                return result;
            }

            // 2,将有效消息，进行序列化存储到临时文件中
            FileHelper::createFile(_tmpfile);
            for (auto &msg : result)
            {
                DLOG("向临时文件写入数据: %s", msg->payload().body().c_str());
                ret = insert(_tmpfile, msg);  //这里会更新msg在文件中的偏移量的
                if (ret == false)
                {
                    DLOG("向临时文件写入消息数据失败");
                    return result;
                }
            }

            // 3,删除原来保存数据的文件
            ret = FileHelper::removeFile(_datafile);
            if (ret == false)
            {
                DLOG("删除源文件失败了");
                return result;
            }

            // 4,修改临时文件名，为源文件名称
            ret = FileHelper(_tmpfile).rename(_datafile);
            if (ret == false)
            {
                DLOG("修改临时文件名称失败");
                return result;
            }

            return result;
        }

        // 加载出文件中所有的有效数据 , 存储的格式是 4字节长度|数据|4字节长度|数据
        bool load(std::list<MessagePtr> &result)
        {
            FileHelper data_file_helper(_datafile);
            size_t offset = 0, msg_size;
            size_t fsize = data_file_helper.size(); // 数据文件的大小
            bool ret;
            while (offset < fsize)
            {
                ret = data_file_helper.read((char *)&msg_size, offset, sizeof(size_t));
                if (ret == false)
                {
                    DLOG("读取消息长度失败");
                    return false;
                }
                offset += sizeof(size_t);
                std::string msg_body(msg_size, '\0');
                ret = data_file_helper.read(&msg_body[0], offset, msg_size);
                if (ret == false)
                {
                    DLOG("读取消息数据失败");
                    return false;
                }
                offset += msg_size;
                MessagePtr msgp = std::make_shared<Message>();
                msgp->mutable_payload()->ParseFromString(msg_body);
                if (msgp->payload().valid() == "0")
                {
                    DLOG("这个是无效数据: %s", msgp->payload().body().c_str());
                    continue;
                }
                result.push_back(msgp);
            }

            return true;
        }

    private:
        std::string _qname;
        std::string _datafile;
        std::string _tmpfile;
    };

    // 消息以队列为单元进行管理
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string &basedir, const std::string &qname)
            : _mapper(basedir, qname), _qname(qname), _valid_count(0), _total_count(0)
        {}

        bool recovery()
        {
            // 恢复历史消息
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.gc();
            for (auto &msg : _msgs)
            {
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }

            _valid_count = _total_count = _msgs.size();
            return true;
        }

        bool insert(const BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            // 构造消息对象
            MessagePtr msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);
            if (bp != nullptr)
            {
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }

            std::unique_lock<std::mutex> lock(_mutex);
            // 判断消息是否需要持久化
            if (msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                DLOG("持久化存储消息: %s", body.c_str());
                msg->mutable_payload()->set_valid("1"); // 在持久化存储中表示
                // 持久化存储
                bool ret = _mapper.insert(msg); // 写入文件中
                if (ret == false)
                {
                    DLOG("持久化存储消息: %s 失败了", body.c_str());
                    return false;
                }

                _valid_count += 1;
                _total_count += 1;
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }

            _msgs.push_back(msg);
            return true;
        }

        // 每次删除消息以后，主要是从待确认消息集合中删除,判断是否需要垃圾回收
        bool remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 1,从待确认队列中查找信息
            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end())
            {
                DLOG("没有找到要删除的消息: %s", msg_id.c_str());
                return true;
            }
            // 2,根据消息的持久化模式，决定是否删除持久化消息
            if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                _mapper.remove(it->second);
                _durable_msgs.erase(msg_id);
                _valid_count -= 1;
                gc(); // 内部判断是否需要垃圾回收，需要的话则回收一下
            }

            _waitack_msgs.erase(msg_id);
            return true;
        }

        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }

        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }

        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }

        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }

        void gc()
        {
            //进行垃圾回收，获取到垃圾回收以后,有效的消息信息链表
            if(GCCheck() == false)
                return;
            
            std::list<MessagePtr> msgs = _mapper.gc();
            for(auto& msg: msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if(it == _durable_msgs.end())
                {
                    DLOG("垃圾回收后,有一条持久化消息，在内存中没有进行管理");
                    _msgs.push_back(msg);
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                    continue;
                }

                //更新每一条消息的实际存储位置
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }

            //更新当前的有效消息数量，总的持久化消息数量
            _valid_count = _total_count = msgs.size();
        }

        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_msgs.size() == 0)
            {
                return MessagePtr();
            }

            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            // 将这个消息对象，向待确认的hash表中添加一份，等收到确认消息以后再删除
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }


    private:
        bool GCCheck()
        {
            //持久化的消息总量大于2000, 且其中有效比例低于50%则需要持久化
            if (_total_count > 2000 && _valid_count * 10  / _total_count < 5) 
            {
                return true;
            }
            return false;
        }

    private:
        std::mutex _mutex;
        std::string _qname;          // 队列名
        size_t _valid_count;         // 持久化的有效消息数量
        size_t _total_count;         // 总的消息数量
        MessageMapper _mapper;       // 消息持久化管理类
        std::list<MessagePtr> _msgs; // 待推送消息链表
        // 持久化hash,
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息hash
        // 待确认消息hash,消息推送给客户端以后，会等待客户端进行消息确认，收到确认以后，才会真正的删除信息
        std::unordered_map<std::string, MessagePtr> _waitack_msgs;
    };

    //实现一个对外的总体消息管理类
    //管理的是每一个队列的消息
    //管理的成员：每个队列的消息管理句柄: 互斥锁 && 队列名称 && 队列消息管理句柄的hash表
    //提供的操作：
    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;
        MessageManager(const std::string& basedir)
            :_basedir(basedir)
        {
            // 其余的在构造函数可以恢复历史消息数据
            // 但是对于消息message而言，只有主动调用initQueueMessage才可以恢复
        }

        //删除所有的消息
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for(auto & qmsg : _queue_msgs)
            {
                qmsg.second->clear();
            }
        }

        void initQueueMessage(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it != _queue_msgs.end())
                    return; //找到了就返回

                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qmp)); 
            }
            qmp->recovery(); //恢复历史数据
        }

        void destroyQueueMessage(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                    return;
                
                qmp = it->second;
                _queue_msgs.erase(it);
            }
            qmp->clear();
        }

        bool insert(const std::string& qname, BasicProperties* bp, const std::string& body, bool queue_is_durable)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("向队列新增消息失败,没有找到消息管理句柄: %s", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            return qmp->insert(bp, body, queue_is_durable);
        }

        //获取队列的队首消息
        MessagePtr front(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列的队首消息失败 %s", qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }

            return qmp->front();
        }

        //确认消息后，从队列里面移除
        void ack(const std::string& qname, const std::string& msg_id)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("确认队列%s消息%s失败：没有找到消息管理句柄!", qname.c_str(), msg_id.c_str());
                    return ;
                }
                qmp = it->second;
            }

            qmp->remove(msg_id);
            return ;
        }

        //获得一个队列中的所有消息数量
        size_t getable_count(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) 
                {
                    DLOG("获取队列%s待推送消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();  
        }

        //返回的是持久化的消息的数量
        size_t durable_count(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s有效持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }

            return qmp->durable_count();
        }

        //待确认的消息数量
        size_t waitack_count(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s待确认消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }

            return qmp->waitack_count();
        }
    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs; //队列名称和队列对象的一个映射关系
    };
}

#endif